package com.gaff.emp.core.hbase;

import cn.hutool.core.date.format.FastDateFormat;
import cn.hutool.core.util.StrUtil;
import com.gaff.emp.core.EmpException;
import com.gaff.emp.core.hbase.desc.*;
import com.gaff.emp.core.util.Identities;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Slf4j
public class HBaseAdminTemplate extends AbstractHBaseTemplate {

    public static final String TABLE_NAME_SPLIT_CHAR = ":";

    public static final String DEFAULT_NAMESPACE_NAME = "default";

    public static final Integer DISABLE_REPLICATION_SCOPE = 0;

    public static final Integer ENABLE_REPLICATION_SCOPE = 1;

    public HBaseAdminTemplate(Configuration configuration) {
        super(configuration);
    }

    public HBaseAdminTemplate(String quorum) {
        super(quorum);
    }

    /**
     * 判断HBase表是否存在
     *
     * @param tableName HBase表名
     * @return HBase表是否存在
     */
    public boolean tableExists(String tableName) {
        return this.execute(admin -> admin.tableExists(TableName.valueOf(tableName)));
    }

    /**
     * 获取所有的HBase表及其描述
     *
     * @return 所有的HBase表及其描述
     */
    public List<HBaseTableDesc> listTableDesc() {
        return listTableDesc("", false);
    }

    /**
     * @param includeSysTables 是否包含系统表
     * @return 所有的HBase表及其描述
     */
    public List<HBaseTableDesc> listTableDesc(boolean includeSysTables) {
        return listTableDesc("", includeSysTables);
    }

    /**
     * 正则查询HBase表及其描述
     *
     * @param regex            查询正则
     * @param includeSysTables 是否包含系统表
     * @return 筛选出的HBase表及其描述
     */
    public List<HBaseTableDesc> listTableDesc(String regex, boolean includeSysTables) {
        return this.execute(admin -> {
            List<TableDescriptor> tableDescriptors;
            if (StrUtil.isBlank(regex)) {
                tableDescriptors = admin.listTableDescriptors(null, includeSysTables);
            } else {
                tableDescriptors = admin.listTableDescriptors(Pattern.compile(regex), includeSysTables);
            }
            if (tableDescriptors == null || tableDescriptors.isEmpty()) {
                return new ArrayList<>();
            }

            List<HBaseTableDesc> tableDescList = parseHBaseTableDescriptorsToTableDescList(tableDescriptors);

            for (HBaseTableDesc tableDesc : tableDescList) {
                tableDesc.setLastMajorCompactTimestamp(admin.getLastMajorCompactionTimestamp(TableName.valueOf(tableDesc.getTableName())));
            }
            return tableDescList;
        });
    }

    /**
     * 获取某一命名空间下的所有表信息
     *
     * @param namespaceName 命名空间名称
     * @return 所有表信息
     */
    public List<HBaseTableDesc> listTableDescByNamespace(String namespaceName) {
        return this.execute(admin -> {
            final List<TableDescriptor> tableDescriptors = admin.listTableDescriptorsByNamespace(Bytes.toBytes(namespaceName));
            if (tableDescriptors == null || tableDescriptors.isEmpty()) {
                return new ArrayList<>();
            }

            final List<HBaseTableDesc> tableDescList = parseHBaseTableDescriptorsToTableDescList(tableDescriptors);

            for (HBaseTableDesc tableDesc : tableDescList) {
                tableDesc.setLastMajorCompactTimestamp(admin.getLastMajorCompactionTimestamp(TableName.valueOf(tableDesc.getTableName())));
            }
            return tableDescList;
        });
    }

    /**
     * 获取所有表名
     *
     * @return 所有表名
     */
    public List<String> listTableNames() {
        return listTableNames("", false);
    }

    /**
     * 获取所有表名
     *
     * @param includeSysTables 是否包含系统表
     * @return 所有表名
     */
    public List<String> listTableNames(boolean includeSysTables) {
        return listTableNames("", includeSysTables);
    }

    /**
     * 获取所有表名
     *
     * @param regex            查询正则
     * @param includeSysTables 是否包含系统表
     * @return 所有表名
     */
    public List<String> listTableNames(String regex, boolean includeSysTables) {
        return this.execute(admin -> {
            TableName[] tableNames;
            if (StrUtil.isBlank(regex)) {
                tableNames = admin.listTableNames((Pattern) null, includeSysTables);
            } else {
                tableNames = admin.listTableNames(Pattern.compile(regex), includeSysTables);
            }
            if (tableNames == null || tableNames.length == 0) {
                return new ArrayList<>();
            }
            return Arrays.stream(tableNames).map(TableName::getNameAsString).collect(Collectors.toList());
        });
    }

    /**
     * 获取某一命名空间下的所有表名
     *
     * @param namespaceName 命名空间名称
     * @return 所有表名
     */
    public List<String> listTableNamesByNamespace(String namespaceName) {
        List<HBaseTableDesc> tableDescList = listTableDescByNamespace(namespaceName);
        return tableDescList.stream().map(HBaseTableDesc::getTableName).collect(Collectors.toList());
    }

    /**
     * 获取某一张表的描述信息
     *
     * @param tableName 表名
     * @return 表描述
     */
    public HBaseTableDesc getTableDesc(String tableName) {
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));
            HBaseTableDesc tableDesc = parseHBaseTableDescriptorToTableDesc(tableDescriptor);
            tableDesc.setLastMajorCompactTimestamp(admin.getLastMajorCompactionTimestamp(TableName.valueOf(tableName)));
            return tableDesc;
        });
    }

    /**
     * 获取某张表所有的列簇信息
     *
     * @return 所有的列簇信息
     */
    public List<HBaseColumnFamilyDesc> listColumnFamilyDesc(String tableName) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));
            final ColumnFamilyDescriptor[] families = tableDescriptor.getColumnFamilies();
            return parseColumnFamilyDescriptorToFamilyDescList(families);
        });
    }

    /**
     * 创建表，以默认的方式
     *
     * @param tableDesc 表的描述信息
     * @return 表是否被创建成功
     */
    public boolean createTable(HBaseTableDesc tableDesc) {
        tableIsExistsError(tableDesc.getTableName());
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = parseTableDescToHBaseTableDescriptor(tableDesc);
            admin.createTable(tableDescriptor);
            return true;
        });
    }

    /**
     * 创建表，预分区
     *
     * @param tableDesc  表的描述信息
     * @param startKey   预分区开始的key
     * @param endKey     预分区结束的key
     * @param numRegions 需要的预分区个数
     * @param isAsync    是否是异步的方式
     * @return 表是否被创建成功
     */
    public boolean createTable(HBaseTableDesc tableDesc, String startKey, String endKey, int numRegions, boolean isAsync) {
        tableIsExistsError(tableDesc.getTableName());
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = parseTableDescToHBaseTableDescriptor(tableDesc);
            boolean preSplit = (StrUtil.isNotBlank(startKey) && StrUtil.isNotBlank(endKey) && numRegions > 0);

            if (preSplit) {
                final byte[] startKeyBytes = Bytes.toBytes(startKey);
                final byte[] endKeyBytes = Bytes.toBytes(endKey);

                if (numRegions < 3) {
                    throw new EmpException("请至少指定3个分区");
                }

                if (Bytes.compareTo(startKeyBytes, endKeyBytes) >= 0) {
                    throw new EmpException("预分区开始的key必须要小于预分区结束的key");
                }

                if (numRegions == 3) {
                    if (isAsync) {
                        admin.createTableAsync(tableDescriptor, new byte[][]{startKeyBytes, endKeyBytes});
                    } else {
                        admin.createTable(tableDescriptor, new byte[][]{startKeyBytes, endKeyBytes});
                    }
                    return true;
                }
                byte[][] splitKeys = Bytes.split(startKeyBytes, endKeyBytes, numRegions - 3);

                if (splitKeys == null || splitKeys.length != numRegions - 1) {
                    throw new EmpException("无法将预分区的起止键范围分割成足够的region");
                }

                if (isAsync) {
                    admin.createTableAsync(tableDescriptor, splitKeys);
                } else {
                    admin.createTable(tableDescriptor, splitKeys);
                }

            } else {
                admin.createTable(tableDescriptor);
            }
            return true;
        });
    }

    /**
     * 创建表，预分区
     *
     * @param tableDesc 表的描述信息
     * @param splitKeys 需要划分的预分区key
     * @param isAsync   是否是异步的方式
     * @return 表是否被创建成功
     */
    public boolean createTable(HBaseTableDesc tableDesc, String[] splitKeys, boolean isAsync) {
        tableIsExistsError(tableDesc.getTableName());
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = parseTableDescToHBaseTableDescriptor(tableDesc);
            boolean preSplit = splitKeys != null && splitKeys.length > 0;

            if (preSplit) {
                if (isAsync) {
                    admin.createTableAsync(tableDescriptor, getSplitKeyBytes(splitKeys));
                } else {
                    admin.createTable(tableDescriptor, getSplitKeyBytes(splitKeys));
                }
            } else {
                admin.createTable(tableDescriptor);
            }
            return true;
        });
    }

    /**
     * 修改表
     *
     * @param tableDesc 表描述
     * @return 表是否被修改成功
     */
    public boolean modifyTableProps(HBaseTableDesc tableDesc) {
        tableIsNotExistsError(tableDesc.getTableName());
        return this.execute(admin -> {
            final TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableDesc.getTableName()));
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableDescriptor);

            final Map<String, String> tableProps = tableDesc.getTableProps();
            if (tableProps != null && !tableProps.isEmpty()) {
                tableProps.forEach(tableDescriptorBuilder::setValue);
                admin.modifyTable(tableDescriptorBuilder.build());
                return true;
            }
            return true;
        });
    }

    /**
     * 修改表名
     *
     * @param oldTableName   旧表名
     * @param newTableName   新表名
     * @param deleteOldTable 是否删除旧表
     * @return 修改表名结果
     */
    public boolean renameTable(String oldTableName, String newTableName, boolean deleteOldTable) {
        tableIsNotExistsError(oldTableName);
        tableIsExistsError(newTableName);

        if (isTableEnabled(oldTableName)) {
            throw new EmpException("表[" + oldTableName + "]重命名之前请先禁用");
        }

        return this.execute(admin -> {
            final String tempSnapShotName = oldTableName.replace(":", "_") +
                    "_SNAPSHOT_RENAME_" + Identities.uuid();
            admin.snapshot(tempSnapShotName, TableName.valueOf(oldTableName));
            admin.cloneSnapshot(tempSnapShotName, TableName.valueOf(newTableName));
            admin.deleteSnapshot(tempSnapShotName);
            if (deleteOldTable) {
                admin.deleteTable(TableName.valueOf(oldTableName));
            } else {
                admin.enableTableAsync(TableName.valueOf(oldTableName));
            }
            return true;
        });
    }

    /**
     * 删除表
     *
     * @param tableName 表名
     * @return 表是否被删除成功
     */
    public boolean deleteTable(String tableName) {
        return this.execute(admin -> {
            if (!admin.isTableDisabled(TableName.valueOf(tableName))) {
                throw new EmpException("表[" + tableName + "]删除之前请先禁用");
            }

            admin.deleteTable(TableName.valueOf(tableName));
            return true;
        });
    }

    /**
     * 清空表
     *
     * @param tableName      表名
     * @param preserveSplits 是否保留预分区信息
     * @return 表是否被成功清空
     */
    public boolean truncateTable(String tableName, boolean preserveSplits) {
        return this.execute(admin -> {
            if (!admin.isTableDisabled(TableName.valueOf(tableName))) {
                throw new EmpException("表[" + tableName + "]清空之前请先禁用");
            }
            admin.truncateTable(TableName.valueOf(tableName), preserveSplits);
            return true;
        });
    }

    /**
     * 启用表
     *
     * @param tableName 表名
     * @param isAsync   是否是异步的
     * @return 启用表是否成功
     */
    public boolean enableTable(String tableName, boolean isAsync) {
        return this.execute(admin -> {
            if (admin.isTableEnabled(TableName.valueOf(tableName))) {
                return true;
            }
            if (isAsync) {
                admin.enableTableAsync(TableName.valueOf(tableName));
            } else {
                admin.enableTable(TableName.valueOf(tableName));
            }
            return true;
        });
    }

    /**
     * 禁用表
     *
     * @param tableName 表名
     * @param isAsync   是否是异步的
     * @return 禁用表是否成功
     */
    public boolean disableTable(String tableName, boolean isAsync) {
        return this.execute(admin -> {
            if (admin.isTableDisabled(TableName.valueOf(tableName))) {
                return true;
            }
            if (isAsync) {
                admin.disableTableAsync(TableName.valueOf(tableName));
            } else {
                admin.disableTable(TableName.valueOf(tableName));
            }
            return true;
        });
    }

    /**
     * 表是否被启用
     *
     * @param tableName 表名
     * @return 表是否被启用
     */
    public boolean isTableDisabled(String tableName) {
        return this.execute(admin -> admin.isTableDisabled(TableName.valueOf(tableName)));
    }

    /**
     * 表是否被禁用
     *
     * @param tableName 表名
     * @return 表是否被禁用
     */
    public boolean isTableEnabled(String tableName) {
        return this.execute(admin -> admin.isTableEnabled(TableName.valueOf(tableName)));
    }

    /**
     * 表是否可用，即该表所有的region是否可用
     *
     * @param tableName 表名
     * @return 表是否可用
     */
    public boolean isTableAvailable(String tableName) {
        return this.execute(admin -> admin.isTableAvailable(TableName.valueOf(tableName)));
    }

    /**
     * 为某张表新增一个列簇
     *
     * @param tableName  表名
     * @param familyDesc 列簇定义信息
     * @return 新增列簇是否成功
     */
    public boolean addFamily(String tableName, HBaseColumnFamilyDesc familyDesc) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> {
            final TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));
            if (tableDescriptor.hasColumnFamily(Bytes.toBytes(familyDesc.getFamilyName()))) {
                throw new EmpException("ColumnFamily[" + familyDesc.getFamilyName() + "]在表[" + tableName + "]中已经存在");
            }
            ColumnFamilyDescriptor columnDescriptor = parseColumnFamilyDescToColumnFamilyDescriptor(familyDesc);
            admin.addColumnFamily(TableName.valueOf(tableName), columnDescriptor);
            return true;
        });
    }

    /**
     * 删除一个列簇
     *
     * @param tableName        表名
     * @param columnFamilyName 列簇名
     * @return 删除列簇是否成功
     */
    public boolean deleteFamily(String tableName, String columnFamilyName) {
        return this.execute(admin -> {
            final TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));
            if (!tableDescriptor.hasColumnFamily(Bytes.toBytes(columnFamilyName))) {
                throw new EmpException("ColumnFamily[" + columnFamilyName + "]在表[" + tableName + "]中不存在");
            }
            admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(columnFamilyName));
            return true;
        });
    }

    /**
     * 修改一个列簇
     *
     * @param tableName  表名
     * @param familyDesc 列簇描述
     * @return 修改列簇是否成功
     */
    public boolean modifyFamily(String tableName, HBaseColumnFamilyDesc familyDesc) {
        tableIsNotExistsError(tableName);

        return this.execute(admin -> {
            final TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));

            if (!tableDescriptor.hasColumnFamily(Bytes.toBytes(familyDesc.getFamilyName()))) {
                throw new EmpException("待修改ColumnFamily[" + familyDesc.getFamilyName() + "]不存在");
            }

            ColumnFamilyDescriptor columnDescriptor = tableDescriptor.getColumnFamily(Bytes.toBytes(familyDesc.getFamilyName()));
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(columnDescriptor);
            boolean change = false;

            if (columnDescriptor.getMaxVersions() != familyDesc.getMaxVersions()) {
                columnFamilyDescriptorBuilder.setMaxVersions(familyDesc.getMaxVersions());
                change = true;
            }
            if (columnDescriptor.getTimeToLive() != familyDesc.getTimeToLive()) {
                columnFamilyDescriptorBuilder.setTimeToLive(familyDesc.getTimeToLive());
                change = true;
            }
            if (!columnDescriptor.getCompressionType().getName().equalsIgnoreCase(familyDesc.getCompressionType())) {
                columnFamilyDescriptorBuilder.setCompressionType(Compression.Algorithm.valueOf(familyDesc.getCompressionType()));
                change = true;
            }
            if (columnDescriptor.getScope() != familyDesc.getReplicationScope()) {
                columnFamilyDescriptorBuilder.setScope(familyDesc.getReplicationScope());
                change = true;
            }
            if (change) {
                admin.modifyColumnFamily(TableName.valueOf(tableName), columnFamilyDescriptorBuilder.build());
            }
            return true;
        });
    }

    /**
     * 启用replication
     *
     * @param tableName 表名
     * @param families  列簇名
     * @return 启用replication是否成功
     */
    public boolean enableReplicationScope(String tableName, List<String> families) {
        tableIsNotExistsError(tableName);
        return modifyTableReplicationScope(tableName, ENABLE_REPLICATION_SCOPE, families);
    }

    /**
     * 禁用replication
     *
     * @param tableName 表名
     * @param families  列簇名
     * @return 禁用replication是否成功
     */
    public boolean disableReplicationScope(String tableName, List<String> families) {
        tableIsNotExistsError(tableName);
        return modifyTableReplicationScope(tableName, DISABLE_REPLICATION_SCOPE, families);
    }

    /**
     * 刷新表，异步操作
     *
     * @param tableName 表名
     * @return 刷新表命令是否执行成功
     */
    public boolean flush(String tableName) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> {
            admin.flush(TableName.valueOf(tableName));
            return true;
        });
    }

    /**
     * compact操作
     *
     * @param tableName 表名
     * @return compact命令是否执行成功
     */
    public boolean compact(String tableName) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> {
            admin.compact(TableName.valueOf(tableName));
            return true;
        });
    }

    /**
     * major compact操作
     *
     * @param tableName 表名
     * @return major compact命令是否执行成功
     */
    public boolean majorCompact(String tableName) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> {
            admin.majorCompact(TableName.valueOf(tableName));
            return true;
        });
    }

    /**
     * 创建一个命名空间
     *
     * @param namespaceDesc 该命名空间的描述
     * @return namespace是否创建成功
     */
    public boolean createNamespace(HBaseNamespaceDesc namespaceDesc) {
        if (namespaceIsExists(namespaceDesc.getNamespaceName())) {
            throw new EmpException("Namespace[" + namespaceDesc.getNamespaceName() + "]已经存在");
        }

        return this.execute(admin -> {
            NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespaceDesc.getNamespaceName()).build();
            final Map<String, String> namespaceProps = namespaceDesc.getNamespaceProps();
            if (namespaceProps != null && !namespaceProps.isEmpty()) {
                namespaceProps.forEach(namespaceDescriptor::setConfiguration);
            }
            admin.createNamespace(namespaceDescriptor);
            return true;
        });
    }

    /**
     * 判断命名空间是否存在
     *
     * @param namespaceName 命名空间名称
     * @return 是否存在
     */
    public boolean namespaceIsExists(String namespaceName) {
        final List<String> namespaces = listNamespaceNames();
        if (namespaces == null || namespaces.isEmpty()) {
            return false;
        }
        return namespaces.contains(namespaceName);
    }

    /**
     * 删除命名空间
     *
     * @param namespaceName 命名空间名称
     * @return namespace是否删除成功
     */
    public boolean deleteNamespace(String namespaceName) {
        return this.execute(admin -> {
            final List<TableDescriptor> tableDescriptors = admin.listTableDescriptorsByNamespace(Bytes.toBytes(namespaceName));
            if (tableDescriptors != null && !tableDescriptors.isEmpty()) {
                throw new EmpException("Namespace[" + namespaceName + "]下存在表，不能被删除");
            }
            admin.deleteNamespace(namespaceName);
            return true;
        });
    }

    /**
     * 获取一个命名空间的描述
     *
     * @param namespaceName 命名空间的名称
     * @return 该命名空间的描述
     */
    public HBaseNamespaceDesc getNamespaceDesc(String namespaceName) {
        return this.execute(admin -> {
            final NamespaceDescriptor namespaceDescriptor = admin.getNamespaceDescriptor(namespaceName);
            HBaseNamespaceDesc namespaceDesc = new HBaseNamespaceDesc();
            namespaceDesc.setNamespaceName(namespaceDescriptor.getName());
            namespaceDesc.setNamespaceProps(namespaceDescriptor.getConfiguration());
            return namespaceDesc;
        });
    }

    /**
     * 获取HBase所有的命名空间名称
     *
     * @return 所有的命名空间名称
     */
    public List<String> listNamespaceNames() {
        return listNamespaceDesc().stream().map(HBaseNamespaceDesc::getNamespaceName).collect(Collectors.toList());
    }

    /**
     * 获取HBase所有命名空间的描述
     *
     * @return 所有命名空间的描述
     */
    public List<HBaseNamespaceDesc> listNamespaceDesc() {
        return this.execute(admin -> Arrays.stream(admin.listNamespaceDescriptors()).map(namespaceDescriptor -> {
            HBaseNamespaceDesc namespaceDesc = new HBaseNamespaceDesc();
            namespaceDesc.setNamespaceName(namespaceDescriptor.getName());
            namespaceDesc.setNamespaceProps(namespaceDescriptor.getConfiguration());
            return namespaceDesc;
        }).collect(Collectors.toList()));
    }

    /**
     * 获取某张表最后一次的major compact时间戳，如果是0则最新的HFile无法被找到
     *
     * @param tableName 表名
     * @return 时间戳
     */
    public long getLastMajorCompactionTimestamp(String tableName) {
        tableIsNotExistsError(tableName);
        return this.execute(admin -> admin.getLastMajorCompactionTimestamp(TableName.valueOf(tableName)));
    }

    /**
     * 获取某一个region最后major compaction的时间戳
     *
     * @param regionName 表名
     * @return 时间戳
     */
    public long getLastMajorCompactionTimestampForRegion(String regionName) {
        return this.execute(admin -> admin.getLastMajorCompactionTimestampForRegion(Bytes.toBytes(regionName)));
    }

    /**
     * 查询所有快照
     *
     * @return 所有快照
     */
    public List<HBaseSnapshotDesc> listSnapshots() {
        return this.execute(admin -> {
            final List<SnapshotDescription> listSnapshots = admin.listSnapshots();
            if (listSnapshots == null || listSnapshots.isEmpty()) {
                return new ArrayList<>();
            } else {
                return listSnapshots.stream().map(snapshotDescription -> {
                    HBaseSnapshotDesc snapshotDesc = new HBaseSnapshotDesc();
                    snapshotDesc.setSnapshotName(snapshotDescription.getName());
                    snapshotDesc.setTableName(snapshotDescription.getTableName().getNameAsString());
                    snapshotDesc.setCreateTime(snapshotDescription.getCreationTime());
                    return snapshotDesc;
                }).collect(Collectors.toList());
            }
        });
    }

    /**
     * 创建快照
     *
     * @param snapshotDesc 快照信息描述
     * @return 创建快照是否成功
     */
    public boolean snapshot(HBaseSnapshotDesc snapshotDesc) {
        tableIsNotExistsError(snapshotDesc.getTableName());
        return this.execute(admin -> {
            admin.snapshot(snapshotDesc.getSnapshotName(), TableName.valueOf(snapshotDesc.getTableName()));
            return true;
        });
    }

    /**
     * 恢复快照
     *
     * @param snapshotName 快照名称
     * @return 恢复快照是否成功
     */
    public boolean restoreSnapshot(String snapshotName) {
        return this.execute(admin -> {
            admin.restoreSnapshot(snapshotName);
            return true;
        });
    }

    /**
     * 克隆快照
     *
     * @param snapshotName 快照名称
     * @param tableName    表名
     * @return 克隆快照是否成功
     */
    public boolean cloneSnapshot(String snapshotName, String tableName) {
        tableIsExistsError(tableName);
        return this.execute(admin -> {
            admin.cloneSnapshot(snapshotName, TableName.valueOf(tableName));
            return true;
        });
    }

    /**
     * 删除快照
     *
     * @param snapshotName 快照名称
     * @return 删除快照是否成功
     */
    public boolean deleteSnapshot(String snapshotName) {
        return this.execute(admin -> {
            admin.deleteSnapshot(snapshotName);
            return true;
        });
    }

    /**
     * 根据正则批量删除快照
     *
     * @param regex 正则
     * @return 删除快照是否成功
     */
    public boolean deleteSnapshots(String regex) {
        return this.execute(admin -> {
            admin.deleteSnapshot(regex);
            return true;
        });
    }

    /**
     * 刷新汇总指标
     *
     * @return 汇总指标
     */
    public HBaseSummary refreshSummary() {
        return this.execute(admin -> {
            final ClusterMetrics clusterMetrics = admin.getClusterMetrics();

            String currentTime = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis());

            String version = clusterMetrics.getHBaseVersion();
            String clusterId = clusterMetrics.getClusterId();

            int liveServers = clusterMetrics.getLiveServerMetrics().size();
            int deadServers = clusterMetrics.getDeadServerNames().size();
            int regionCount = clusterMetrics.getRegionCount();
            int ritCount = clusterMetrics.getRegionStatesInTransition().size();
            int namespaceCount = admin.listNamespaceDescriptors().length;
            int tableCount = admin.listTableNames().length;
            int snapshotCount = admin.listSnapshots().size();

            double averageLoad = clusterMetrics.getAverageLoad();

            long aggregateRequestPerSecond = clusterMetrics.getLiveServerMetrics().values().stream()
                    .mapToLong(ServerMetrics::getRequestCountPerSecond).sum();

            return new HBaseSummary(currentTime, version, clusterId, liveServers + deadServers,
                    liveServers, deadServers, namespaceCount, tableCount, snapshotCount, regionCount, ritCount, averageLoad, aggregateRequestPerSecond);
        });
    }

    private boolean modifyTableReplicationScope(String tableName, int replicationScope, List<String> families) {
        tableIsNotExistsError(tableName);

        if (families == null || families.isEmpty()) {
            throw new EmpException("请传入一个或多个待修改的ColumnFamily");
        }

        return this.execute(admin -> {
            TableDescriptor tableDescriptor = admin.getDescriptor(TableName.valueOf(tableName));
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableDescriptor);

            for (String family : families) {
                final ColumnFamilyDescriptor columnFamily = tableDescriptor.getColumnFamily(Bytes.toBytes(family));
                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(columnFamily);
                final int currentScope = columnFamily.getScope();
                if (currentScope != replicationScope) {
                    columnFamilyDescriptorBuilder.setScope(replicationScope);
                    tableDescriptorBuilder.modifyColumnFamily(columnFamilyDescriptorBuilder.build());
                }
            }
            return true;
        });
    }

    private byte[][] getSplitKeyBytes(String[] splitKeys) {
        final List<byte[]> bytes = Arrays.stream(splitKeys).distinct().map(Bytes::toBytes).collect(Collectors.toList());
        byte[][] splitKeyBytes = new byte[bytes.size()][];
        for (int i = 0; i < bytes.size(); i++) {
            splitKeyBytes[i] = bytes.get(i);
        }
        return splitKeyBytes;
    }

    private void tableIsNotExistsError(String tableName) {
        if (!tableExists(tableName)) {
            throw new EmpException("HBase表[" + tableName + "]不存在");
        }
    }

    private void tableIsExistsError(String tableName) {
        if (tableExists(tableName)) {
            throw new EmpException("HBase表[" + tableName + "]已经存在");
        }
    }

    private TableDescriptor parseTableDescToHBaseTableDescriptor(HBaseTableDesc tableDesc) {
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableDesc.getTableName()));
        List<HBaseColumnFamilyDesc> familyDescList = tableDesc.getColumnFamilyDescList();
        if (familyDescList == null || familyDescList.isEmpty()) {
            throw new EmpException("请为表[" + tableDesc.getTableName() + "]指定一个或多个ColumnFamily");
        }
        final Map<String, Long> familyCountMap = familyDescList.stream().collect(Collectors.groupingBy(HBaseColumnFamilyDesc::getFamilyName, Collectors.counting()));
        familyCountMap.forEach((familyName, count) -> {
            if (count > 1) {
                throw new EmpException("同一张表中的列簇名[" + familyName + "]应该是唯一的");
            }
        });

        if (tableDesc.getTableProps() != null && !tableDesc.getTableProps().isEmpty()) {
            tableDesc.getTableProps().forEach(tableDescriptorBuilder::setValue);
        }

        for (HBaseColumnFamilyDesc familyDesc : familyDescList) {
            tableDescriptorBuilder.setColumnFamily(parseColumnFamilyDescToColumnFamilyDescriptor(familyDesc));
        }

        return tableDescriptorBuilder.build();
    }

    private List<HBaseTableDesc> parseHBaseTableDescriptorsToTableDescList(List<TableDescriptor> tableDescriptors) {
        if (tableDescriptors == null || tableDescriptors.isEmpty()) {
            return new ArrayList<>();
        }
        return tableDescriptors.stream().map(this::parseHBaseTableDescriptorToTableDesc).collect(Collectors.toList());
    }

    private HBaseTableDesc parseHBaseTableDescriptorToTableDesc(TableDescriptor tableDescriptor) {
        HBaseTableDesc tableDesc = new HBaseTableDesc();
        String tableName = tableDescriptor.getTableName().getNameAsString();
        tableDesc.setNamespaceName(getNamespaceName(tableName));
        tableDesc.setTableName(tableName);
        tableDesc.setDisabled(isTableDisabled(tableName));
        tableDesc.setMetaTable(tableDescriptor.isMetaTable());
        tableDesc.setMaxFileSize(tableDescriptor.getMaxFileSize());
        tableDesc.setMemStoreFlushSize(tableDescriptor.getMemStoreFlushSize());
        tableDesc.setRegionReplication(tableDescriptor.getRegionReplication());
        tableDesc.setCompression(tableDescriptor.isCompactionEnabled());
        tableDesc.setNormalization(tableDescriptor.isNormalizationEnabled());
        tableDesc.setReadonly(tableDescriptor.isReadOnly());

        final Map<Bytes, Bytes> values = tableDescriptor.getValues();
        if (values != null && !values.isEmpty()) {
            values.forEach((key, value) -> tableDesc.addProp(Bytes.toString(key.get()), Bytes.toString(value.get())));
        }

        tableDesc.setTableDesc(tableDescriptor.toString());
        tableDesc.setColumnFamilyDescList(parseColumnFamilyDescriptorToFamilyDescList(tableDescriptor.getColumnFamilies()));
        return tableDesc;
    }

    private List<HBaseColumnFamilyDesc> parseColumnFamilyDescriptorToFamilyDescList(ColumnFamilyDescriptor[] families) {
        return Arrays.stream(families).map(family -> {
            HBaseColumnFamilyDesc familyDesc = new HBaseColumnFamilyDesc();
            familyDesc.setFamilyName(family.getNameAsString());
            familyDesc.setMinVersions(family.getMinVersions());
            familyDesc.setMaxVersions(family.getMaxVersions());
            familyDesc.setTimeToLive(family.getTimeToLive());
            familyDesc.setCompressionType(family.getCompressionType().getName());
            familyDesc.setReplicationScope(family.getScope());
            familyDesc.setBlockSize(family.getBlocksize());
            familyDesc.setBlockCache(family.isBlockCacheEnabled());
            familyDesc.setBloomFilter(family.getBloomFilterType().name());
            familyDesc.setDataBlockEncoding(family.getDataBlockEncoding().name());
            MemoryCompactionPolicy memoryCompactionPolicy = family.getInMemoryCompaction();
            if (memoryCompactionPolicy != null) {
                familyDesc.setInMemoryCompaction(memoryCompactionPolicy.name());
            }
            familyDesc.setKeepDeletedCells(family.getKeepDeletedCells().name());
            return familyDesc;
        }).collect(Collectors.toList());
    }

    private ColumnFamilyDescriptor parseColumnFamilyDescToColumnFamilyDescriptor(HBaseColumnFamilyDesc familyDesc) {
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyDesc.getFamilyName()));
        columnFamilyDescriptorBuilder.setMaxVersions(familyDesc.getMaxVersions());
        columnFamilyDescriptorBuilder.setTimeToLive(familyDesc.getTimeToLive());
        columnFamilyDescriptorBuilder.setCompressionType(Compression.Algorithm.valueOf(familyDesc.getCompressionType()));
        return columnFamilyDescriptorBuilder.build();
    }

    private String getNamespaceName(String tableName) {
        if (tableName.contains(TABLE_NAME_SPLIT_CHAR)) {
            return tableName.split(TABLE_NAME_SPLIT_CHAR)[0];
        } else {
            return DEFAULT_NAMESPACE_NAME;
        }
    }

    private <T> T execute(AdminHandler<T> action) {
        Admin admin = null;
        try {
            admin = this.getConnection().getAdmin();
            return action.handle(admin);
        } catch (Throwable throwable) {
            throw new EmpException(throwable.getLocalizedMessage());
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                }
            }
        }
    }

}
